package org.fun.kafkademo.producer;

import org.fun.kafkademo.model.BatchProcessResult;
import org.fun.kafkademo.model.UserEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * 批量数据生产者
 */
@Component
public class BatchProducer {
    
    private static final Logger logger = LoggerFactory.getLogger(BatchProducer.class);
    
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    
    private static final String TOPIC_NAME = "user-events";
    private static final String[] EVENT_TYPES = {"LOGIN", "LOGOUT", "PAGE_VIEW", "CLICK", "PURCHASE", "SEARCH"};
    private static final String[] USER_AGENTS = {
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36",
        "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36"
    };
    private static final String[] IP_ADDRESSES = {
        "192.168.1.100", "192.168.1.101", "192.168.1.102", "10.0.0.1", "10.0.0.2"
    };
    
    private final Random random = new Random();
    
    /**
     * 生成指定数量的用户事件数据
     */
    public List<UserEvent> generateUserEvents(int count) {
        List<UserEvent> events = new ArrayList<>();
        LocalDateTime now = LocalDateTime.now();
        
        for (int i = 0; i < count; i++) {
            UserEvent event = new UserEvent(
                "user_" + (random.nextInt(1000) + 1),
                EVENT_TYPES[random.nextInt(EVENT_TYPES.length)],
                "Event data " + i + " - " + System.currentTimeMillis(),
                "session_" + (random.nextInt(100) + 1),
                now.minusMinutes(random.nextInt(60)),
                IP_ADDRESSES[random.nextInt(IP_ADDRESSES.length)],
                USER_AGENTS[random.nextInt(USER_AGENTS.length)]
            );
            events.add(event);
        }
        
        return events;
    }
    
    /**
     * 批量发送用户事件数据
     */
    public BatchProcessResult sendBatchEvents(List<UserEvent> events) {
        LocalDateTime startTime = LocalDateTime.now();
        int successCount = 0;
        int failureCount = 0;
        List<String> errors = new ArrayList<>();
        
        logger.info("开始批量发送 {} 条用户事件数据", events.size());
        
        List<CompletableFuture<SendResult<String, Object>>> futures = new ArrayList<>();
        
        for (UserEvent event : events) {
            try {
                // 使用用户ID作为分区键，确保同一用户的事件发送到同一分区
                String key = event.getUserId();
                
                ListenableFuture<SendResult<String, Object>> future = 
                    kafkaTemplate.send(TOPIC_NAME, key, event);
                
                CompletableFuture<SendResult<String, Object>> completableFuture = 
                    future.completable();
                
                completableFuture.whenComplete((result, throwable) -> {
                    if (throwable != null) {
                        logger.error("发送消息失败: {}", throwable.getMessage());
                    } else {
                        logger.debug("消息发送成功: partition={}, offset={}", 
                            result.getRecordMetadata().partition(),
                            result.getRecordMetadata().offset());
                    }
                });
                
                futures.add(completableFuture);
                
            } catch (Exception e) {
                failureCount++;
                errors.add("发送事件失败: " + e.getMessage());
                logger.error("发送事件失败", e);
            }
        }
        
        // 等待所有发送完成
        for (CompletableFuture<SendResult<String, Object>> future : futures) {
            try {
                future.get(30, TimeUnit.SECONDS);
                successCount++;
            } catch (InterruptedException | ExecutionException | TimeoutException e) {
                failureCount++;
                errors.add("等待发送结果超时或失败: " + e.getMessage());
                logger.error("等待发送结果失败", e);
            }
        }
        
        LocalDateTime endTime = LocalDateTime.now();
        
        BatchProcessResult result = new BatchProcessResult(
            events.size(), successCount, failureCount, errors, startTime, endTime
        );
        
        logger.info("批量发送完成: {}", result);
        return result;
    }
    
    /**
     * 异步批量发送用户事件数据
     */
    public void sendBatchEventsAsync(List<UserEvent> events, 
                                   ListenableFutureCallback<BatchProcessResult> callback) {
        CompletableFuture.supplyAsync(() -> {
            return sendBatchEvents(events);
        }).whenComplete((result, throwable) -> {
            if (throwable != null) {
                callback.onFailure(throwable);
            } else {
                callback.onSuccess(result);
            }
        });
    }
    
    /**
     * 发送单个用户事件
     */
    public void sendSingleEvent(UserEvent event) {
        try {
            String key = event.getUserId();
            kafkaTemplate.send(TOPIC_NAME, key, event)
                .addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
                    @Override
                    public void onSuccess(SendResult<String, Object> result) {
                        logger.info("单个事件发送成功: partition={}, offset={}", 
                            result.getRecordMetadata().partition(),
                            result.getRecordMetadata().offset());
                    }
                    
                    @Override
                    public void onFailure(Throwable ex) {
                        logger.error("单个事件发送失败", ex);
                    }
                });
        } catch (Exception e) {
            logger.error("发送单个事件失败", e);
        }
    }
}
